package lb.io.client;

import com.google.protobuf.ByteString;
import io.grpc.*;
import io.grpc.stub.StreamObserver;
import lb.io.api.system.v1.Component;
import lb.io.apimachinery.pkg.apis.meta.v1.*;
import lb.io.apimachinery.pkg.apis.meta.v1.Status;
import lb.io.apimachinery.runtime.Plant;
import lb.io.apimachinery.runtime.scheme.GroupVersionKind;
import lb.io.apiserver.pkg.service.pb.*;
import lb.io.options.ClientOptions;
import lb.io.utils.Klog;
import org.apache.logging.log4j.Logger;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class APIServerClient {
    enum State {
        Block("block"), Ready("ready");
        String state;

        State(String state) {
            this.state = state;
        }
    }

    // APIServerClient 连接状态
    private static State state = State.Ready;

    private static final Logger klog = Klog.getLogger(APIServerClient.class);

    private final String contentType = "application/vk.linkBackup.protobuf";

    private ClientOptions clientOptions;

    private final Channel channel;

    private List<chatParam> chatParams = new ArrayList<>();

    class chatParam {
        private Component component;

        private StreamObserver<ChatResponse> observer;

        chatParam(Component component, StreamObserver<ChatResponse> observer) {
            this.component = component;
            this.observer = observer;
        }
    }

    public APIServerClient(ClientOptions options) {
        this.clientOptions = options;
        this.channel = ManagedChannelBuilder
                .forTarget(options.getEndpoints())
                .userAgent(options.getName())
                .usePlaintext()
                .keepAliveTime(options.getKeepaliveDuration(), TimeUnit.SECONDS)
                .keepAliveTime(options.getKeepaliveTimeout(), TimeUnit.SECONDS)
                .keepAliveWithoutCalls(true)
                // .maxRetryAttempts(3)
                .executor(Executors.newFixedThreadPool(5))
                .build();

        heart();
    }

    public void heart() {
        new Timer("heart").schedule(new TimerTask() {
            @Override
            public void run() {
                if (getState().equals(APIServerClient.State.Block)) {
                    reconnect();
                    return;
                }

                try {
                    pingz();
                    setState(State.Ready);
                } catch (StatusRuntimeException e) {
                    setState(State.Block);
                }

            }
        }, 0, 5000);
    }

    public void reconnect() {
        Close();
        try {
            pingz();
            for (chatParam param : chatParams) {
                chat(param.component, param.observer);
                klog.info("reconnecting chat stream");
            }
            setState(State.Ready);
        } catch (StatusRuntimeException e) {
            klog.error("reconnect failed: " + e.getMessage());
        }
    }

    public APIServerClient(Channel channel) {
        this.channel = channel;
    }

    private static synchronized State getState() {
        return state;
    }

    private synchronized void setState(State state) {
        this.state = state;
    }

    private APIServerGrpc.APIServerBlockingStub getBlockingStub() {
        return APIServerGrpc.newBlockingStub(channel);
    }

    /**
     * healthz 检测 apiserver 状态
     *
     * @return
     * @throws StatusRuntimeException
     */
    public String healthz() throws StatusRuntimeException {
        HealthzRequest request = HealthzRequest.newBuilder().build();
        return getBlockingStub().healthz(request).getMessage();
    }

    public StreamObserver<ChatResponse> Chat(final Component component, StreamObserver<ChatResponse> responseObserver)
            throws StatusRuntimeException {

        chatParams.add(new chatParam(component, responseObserver));

        return chat(component, responseObserver);
    }

    /**
     * @param component
     * @return
     * @throws StatusRuntimeException
     */
    private StreamObserver<ChatResponse> chat(final Component component, StreamObserver<ChatResponse> responseObserver)
            throws StatusRuntimeException {

        Channel customChan = ClientInterceptors.intercept(channel, new ClientInterceptor() {
            @Override
            public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method,
                                                                       CallOptions callOptions, Channel next) {
                return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
                    @Override
                    public void start(Listener<RespT> responseListener, Metadata headers) {
                        super.start(new ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT>(responseListener) {
                            @Override
                            public void onHeaders(Metadata headers) {
                                super.onHeaders(headers);
                            }
                        }, headers);
                    }
                };
            }
        });
        APIServerGrpc.APIServerStub stub = APIServerGrpc.newStub(customChan);
        ChatRequest request = ChatRequest.newBuilder()
                .setComponent(component)
                .build();

        StreamObserver<ChatRequest> requestObserver = stub.chat(responseObserver);
        requestObserver.onNext(request);
        return responseObserver;
    }

    /**
     * ping 请求 apiserver
     *
     * @return PingOutPut
     * @throws StatusRuntimeException
     */
    public PingOutPut pingz() throws StatusRuntimeException {
        PingzRequest request = PingzRequest.newBuilder().build();
        PingzResponse response = getBlockingStub().pingz(request);
        return PingOutPut.newBuilder()
                .setMessage(response.getOut().getMessage())
                .setRemoteIP(response.getOut().getRemoteIP())
                .build();
    }

    /**
     * 更新 Object 的状态为 ready
     *
     * @param raws
     * @return ObjectRaws
     * @throws StatusRuntimeException
     */
    public ObjectRaws readyz(ObjectRaws raws) throws StatusRuntimeException {
        ReadyzRequest request = ReadyzRequest.newBuilder()
                .setRaws(raws)
                .build();
        ReadyzResponse response = getBlockingStub().readyz(request);
        return response.getRaws();
    }

    /**
     * 修改资源的状态为 readyz
     *
     * @param gvk
     * @param object
     * @param ttl
     * @return
     * @throws StatusRuntimeException
     */
    public Object TryReadyz(GroupVersionKind gvk, Object object, int ttl) throws StatusRuntimeException {
        if (getState().equals(State.Block)) {
            klog.warn("apiserver client block");
            return null;
        }

        Object outObj;
        try {
            outObj = getstate(gvk, object, 3);
        } catch (Exception e) {
            return postz(gvk, object, ttl, false);
        }

        if (outObj == null) {
            return null;
        }

        Plant plant;
        try {
            plant = new Plant(gvk, outObj);
            ObjectRaws raws = plant.raws(contentType, ttl);
            ObjectRaws objectRaws = readyz(raws);
            outObj = plant.setRV(objectRaws.getResourceVersion());
            return parse(outObj, objectRaws.getRaws());
        } catch (Exception e) {
            klog.warn(e.getMessage());
            return null;
        }
    }

    /**
     * 从 APIServer 中获取资源
     *
     * @param gvk
     * @param object
     * @return
     */
    public Object GetObj(GroupVersionKind gvk, Object object) {
        if (getState().equals(State.Block)) {
            klog.warn("apiserver client block");
            return null;
        }

        GetOptions options = null;
        try {
            options = new Plant(gvk, object).getOptions(contentType);
        } catch (Exception e) {
            klog.warn(e.getMessage());
            return null;
        }

        Output output = get(options);
        return parse(object, output.getRaws());
    }

    private Object parse(Object obj, ByteString data) {
        Object out = null;
        try {
            Class c1 = Class.forName(obj.getClass().getName());
            Method m = c1.getDeclaredMethod("parseFrom", ByteString.class);
            m.setAccessible(true);
            out = m.invoke(c1, data);
            ObjectMeta meta = (ObjectMeta) out.getClass().getDeclaredMethod("getMetadata").invoke(out);
            if (meta.getName().equals("")) {
                out = null;
            }
        } catch (Exception e) {
            klog.warn(e.getMessage());
            return null;
        }
        return out;
    }

    /**
     * 获取资源信息，如果资源不存在，直接创建
     *
     * @param gvk
     * @param object
     * @return
     */
    public Object GetOrCreate(GroupVersionKind gvk, Object object) {
        if (getState().equals(State.Block)) {
            klog.warn("apiserver client block");
            return null;
        }
        Object outObj = GetObj(gvk, object);
        if (outObj != null) {
            return outObj;
        }
        return postz(gvk, object, 0, false);
    }

    /**
     * 请求当个资源
     *
     * @param options
     * @return
     * @throws StatusRuntimeException
     */
    public Output get(GetOptions options) throws StatusRuntimeException {
        GetRequest request = GetRequest.newBuilder()
                .setBody(options)
                .build();
        GetResponse response = getBlockingStub().get(request);
        return response.getOut();
    }

    /**
     * 获取资源的最新状态
     *
     * @param gvk
     * @param object
     * @param retry
     * @return
     * @throws StatusRuntimeException
     * @throws NoSuchMethodException
     * @throws IllegalAccessException
     * @throws InvocationTargetException
     * @throws InstantiationException
     */
    private Object getstate(GroupVersionKind gvk, Object object, int retry) throws StatusRuntimeException, NoSuchMethodException, IllegalAccessException, InvocationTargetException {
        int n = 0;
        while (true) {
            n++;
            Plant plant = new Plant(gvk, object);
            GetOptions options = plant.getOptions(contentType);
            Output output = get(options);
            if (output != null) {
                return plant.setRV(output.getResourceVersion());
            }
            if (n >= retry) {
                return null;
            }
        }
    }

    /**
     * 查询资源的集合
     *
     * @param gvk
     * @param object
     * @param limit
     * @param fields    查询条件
     * @param _continue 分页的开始路径
     * @return
     */
    public Object GetList(GroupVersionKind gvk, Object object, int limit, Map<String, String> fields, String _continue) {
        if (getState().equals(State.Block)) {
            klog.warn("apiserver client block");
            return null;
        }

        ListOptions options;
        try {
            options = new Plant(gvk, object).listOptions(contentType, limit, fields, _continue);
        } catch (Exception e) {
            klog.warn(e.getMessage());
            return null;
        }

        Output output = list(options);
        return parse(object, output.getRaws());
    }

    /**
     * 请求资源的集合
     *
     * @param options
     * @return
     * @throws StatusRuntimeException
     */
    public Output list(ListOptions options) throws StatusRuntimeException {
        ListRequest request = ListRequest.newBuilder()
                .setBody(options)
                .build();
        ListResponse response = getBlockingStub().list(request);
        return response.getOut();
    }

    /**
     * 创建一个资源，同时在 APIServer 触发一个事件
     *
     * @param gvk
     * @param object
     * @param ttl
     * @return
     */
    public Object TryFakePost(GroupVersionKind gvk, Object object, int ttl) {
        if (getState().equals(State.Block)) {
            klog.warn("apiserver client block");
            return null;
        }

        return postz(gvk, object, ttl, true);
    }

    /**
     * 直接创建一个资源
     *
     * @param gvk
     * @param object
     * @param ttl    资源的生命周期，单位为秒
     * @return
     */
    public Object TryPost(GroupVersionKind gvk, Object object, int ttl) {
        if (getState().equals(State.Block)) {
            klog.warn("apiserver client block");
            return null;
        }

        return postz(gvk, object, ttl, false);
    }

    private Object postz(GroupVersionKind gvk, Object object, int ttl, boolean fake) {
        PostOptions options;
        try {
            options = new Plant(gvk, object).postOptions(contentType, ttl);
        } catch (Exception e) {
            klog.warn(e.getMessage());
            return null;
        }

        Output output;
        if (fake) {
            output = fakePost(options);
        } else {
            output = post(options);
        }
        return parse(object, output.getRaws());
    }

    /**
     * 新增一个资源
     *
     * @param options
     * @return
     * @throws StatusRuntimeException
     */
    public Output post(PostOptions options) throws StatusRuntimeException {
        PostRequest request = PostRequest.newBuilder()
                .setBody(options)
                .build();
        PostResponse response = getBlockingStub().post(request);
        return response.getOut();
    }

    /**
     * 伪添加一个资源，发起该请求时，创建一个资源的同时，生成一个 Event
     *
     * @param options
     * @return
     * @throws StatusRuntimeException
     */
    public Output fakePost(PostOptions options) throws StatusRuntimeException {
        FakePostRequest request = FakePostRequest.newBuilder()
                .setBody(options)
                .build();
        FakePostResponse response = getBlockingStub().fakePost(request);
        return response.getOut();
    }

    /**
     * 修改一个资源
     *
     * @param options
     * @return
     * @throws StatusRuntimeException
     */
    public Output patch(PatchOptions options) throws StatusRuntimeException {
        PatchRequest request = PatchRequest.newBuilder()
                .setBody(options)
                .build();
        PatchResponse response = getBlockingStub().patch(request);
        return response.getOut();
    }

    private Object patchz(GroupVersionKind gvk, Object object, int ttl) {
        PatchOptions options;
        try {
            options = new Plant(gvk, object).patchOptions(contentType, ttl);
        } catch (Exception e) {
            klog.warn(e.getMessage());
            return null;
        }
        Output output = patch(options);

        return parse(object, output.getRaws());
    }

    /**
     * 更新一个资源，先检查资源 rv
     *
     * @param gvk
     * @param object
     * @param ttl
     * @return
     */
    public Object TryUpdate(GroupVersionKind gvk, Object object, int ttl) {
        if (getState().equals(State.Block)) {
            klog.warn("apiserver client block");
            return null;
        }

        try {
            Object outObj = getstate(gvk, object, 1);
            return patchz(gvk, outObj, ttl);
        } catch (Exception e) {
            klog.warn(e.getMessage());
            return null;
        }
    }

    /**
     * 删除一个资源
     *
     * @param options
     * @return
     * @throws StatusRuntimeException
     */
    public Output delete(DeleteOptions options) throws StatusRuntimeException {
        DeleteRequest request = DeleteRequest.newBuilder()
                .setBody(options)
                .build();
        DeleteResponse response = getBlockingStub().delete(request);
        return response.getOut();
    }

    /**
     * 伪删除一个资源，并发起一个 Event
     *
     * @param options
     * @return
     * @throws StatusRuntimeException
     */
    public Output fakeDelete(DeleteOptions options) throws StatusRuntimeException {
        FakeDeleteRequest request = FakeDeleteRequest.newBuilder()
                .setBody(options)
                .build();
        FakeDeleteResponse response = getBlockingStub().fakeDelete(request);
        return response.getOut();
    }

    private Object deletez(GroupVersionKind gvk, Object object, boolean fake) {
        DeleteOptions options;
        try {
            options = new Plant(gvk, object).deleteOptions(contentType);
        } catch (Exception e) {
            klog.warn(e.getMessage());
            return null;
        }
        Output output;
        if (fake) {
            output = fakeDelete(options);
        } else {
            output = delete(options);
        }

        return parse(object, output.getRaws());
    }

    /**
     * 删除一个资源，会首先更新下资源的 rv
     *
     * @param gvk
     * @param object
     * @return
     */
    public Object TryDelete(GroupVersionKind gvk, Object object) {
        if (getState().equals(State.Block)) {
            klog.warn("apiserver client block");
            return null;
        }

        try {
            Object outObj = getstate(gvk, object, 1);
            return deletez(gvk, outObj, false);
        } catch (Exception e) {
            klog.warn(e.getMessage());
            return null;
        }
    }

    /**
     * 伪删除一个资源，触发一个时间，会首先更新下资源的 rv
     *
     * @param gvk
     * @param object
     * @return
     */
    public Object TryFakeDelete(GroupVersionKind gvk, Object object) {
        if (getState().equals(State.Block)) {
            klog.warn("apiserver client block");
            return null;
        }

        try {
            Object outObj = getstate(gvk, object, 1);
            return deletez(gvk, outObj, true);
        } catch (Exception e) {
            klog.warn(e.getMessage());
            return null;
        }
    }

    /**
     * @param options
     * @return
     * @throws StatusRuntimeException
     */
    public StreamObserver<WatchResponse> watch(final WatchOptions options, StreamObserver<WatchResponse> responseObserver)
            throws StatusRuntimeException {
        Channel customChan = ClientInterceptors.intercept(channel, new ClientInterceptor() {
            @Override
            public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method,
                                                                       CallOptions callOptions, Channel next) {
                return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, callOptions)) {
                    @Override
                    public void start(Listener<RespT> responseListener, Metadata headers) {
                        Metadata.Key<String> uid = Metadata.Key.of("uid", Metadata.ASCII_STRING_MARSHALLER);
                        headers.put(uid, options.getClientID());
                        super.start(new ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT>(responseListener) {
                            @Override
                            public void onHeaders(Metadata headers) {
                                super.onHeaders(headers);
                            }
                        }, headers);
                    }
                };
            }
        });
        APIServerGrpc.APIServerStub stub = APIServerGrpc.newStub(customChan);
        WatchRequest request = WatchRequest.newBuilder()
                .setMessage("+start")
                .setBody(options)
                .build();
        StreamObserver<WatchRequest> requestObserver = stub.watch(responseObserver);
        requestObserver.onNext(request);
        return responseObserver;
    }

    /**
     * @param body
     * @return
     * @throws StatusRuntimeException
     */
    public StreamObserver<HandleResponse> handle(final EventOption body, StreamObserver<HandleResponse> responseObserver)
            throws StatusRuntimeException {
        Channel customChan = ClientInterceptors.intercept(channel, new ClientInterceptor() {
            @Override
            public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method,
                                                                       CallOptions options, Channel next) {
                return new ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT>(next.newCall(method, options)) {
                    @Override
                    public void start(Listener<RespT> responseListener, Metadata headers) {
                        Metadata.Key<String> uid = Metadata.Key.of("uid", Metadata.ASCII_STRING_MARSHALLER);
                        headers.put(uid, body.getClientID());
                    }
                };
            }
        });

        APIServerGrpc.APIServerStub stub = APIServerGrpc.newStub(customChan);
        HandleRequest request = HandleRequest.newBuilder()
                .setMessage("+start")
                .setBody(body)
                .build();
        StreamObserver<HandleRequest> requestObserver = stub.handle(responseObserver);
        requestObserver.onNext(request);
        return responseObserver;
    }

    public void Close() {
        // TODO: APIServerClient Close
    }
}
